package com.exchange.plugin.kafka;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.apache.kafka.common.TopicPartition;
import com.exchange.otherservice.KafkaConsumerService;
import com.jfinal.aop.Inject;

import cn.hutool.json.JSONUtil;
import java.time.Duration;
import java.util.Collections;
import java.util.HashSet;
import java.util.Properties;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Future;



/**
 * kafka工具类
 *
 */
public class Kafka {
	
	//ticp,map-key,group.id 共用名称
    public static final String matchTrade = "matchTrade";
    public static final String orderChange = "orderChange";
    public static final String tradeDepth = "tradeDepth";
    public static final String revokeResult = "revokeResult";
    
    public static final String matchOrder = "matchOrder";
    public static final String revokeOrder = "revokeOrder";
    
	public static final ConcurrentHashMap<String, KafkaProducer<Long, String>> producerMap = new ConcurrentHashMap<String, KafkaProducer<Long, String>>();//生产者
	public static final ConcurrentHashMap<String, KafkaConsumer<Long, String>> consumerMap = new ConcurrentHashMap<String, KafkaConsumer<Long, String>>();//消费者
	
	
	//消费者消费开关
	
	
	/**
     * 添加生产者
     * @param name
     * @param servers
     * @param keySerializer
     * @param valueSerializer
     */
    public static void addProducer(String producerName,Properties props) {
        if (!producerMap.containsKey(producerName)) {
            KafkaProducer<Long, String> producer = new KafkaProducer<Long, String>(props);
            producerMap.put(producerName, producer);
        }
    }
    
    /**
	 * 添加消费者
	 */
    public static KafkaConsumer<Long, String> addConsumer(String consumerName,String topic,Properties props) {
    	props.put("group.id", consumerName);
    	
    	if (consumerMap.containsKey(consumerName)) {
    		return consumerMap.get(consumerName);
    	}else {
    		KafkaConsumer<Long, String> consumer = new KafkaConsumer<Long, String>(props);
	    	consumer.subscribe(Collections.singletonList(topic));//订阅TOPIC
	    	consumerMap.put(consumerName, consumer);
	    	return consumerMap.get(consumerName);
    	}
    }
	
    /**
     * 获取一个消费者
     */
    public static KafkaConsumer<Long, String> getConsumer(String name) {
		return consumerMap.get(name);
	}
	
	/**
	 * 获取一个生产者
	 * @param name
	 * @return
	 */
	public static KafkaProducer<Long, String> getProducer(String name) {
		return producerMap.get(name);
	}
	
	
	/**
	 * 生产者发送消息
	 * @param <T>
	 * @param <T>
	 */
	public static <T> Future<RecordMetadata> send(String producerName,String topicName,Long key,T Obj) {
		String jsonStr = JSONUtil.toJsonStr(Obj);
		return producerMap.get(producerName).send(new ProducerRecord<Long, String>(topicName, key, jsonStr));
	}
	
	
	/**
	 * 消费者消费
	 * @param consumer1
	 */
	public static void matchOrderConsumer(KafkaConsumer<Long, String> consumer1){
    	try {
		    while (true) {
		        // 100 是超时时间（ms），在该时间内 poll 会等待服务器返回数据
		    	ConsumerRecords<Long, String> records = consumer1.poll(Duration.ofMillis(Long.MAX_VALUE)); 
		        // poll 返回一个记录列表。
		        // 每条记录都包含了记录所属主题的信息、记录所在分区的信息、记录在分区里的偏移量，以及记录的键值对。(Trade部分)
		    	//country={"isBuy":true,"uid":455951945836789761,"number":1,"dealWay":"maker","coinTeam":"XBITUSDT","totalPrice":10,"price":10,"oid":455951945836789760}
		        for (ConsumerRecord<Long, String> record : records) {
		        	 KafkaConsumerService.insertValue(record.value());
			    }
		        /*consumer.commitSync();//同步*/
		    } 
		} finally {
		    // 关闭消费者,网络连接和 socket 也会随之关闭，并立即触发一次再均衡
			consumer1.close();
		}
    }
	
	/**
	 * 消费者消费
	 * @param consumer2
	 */
	public static void revokeOrderConsumer(KafkaConsumer<Long, String> consumer2){
    	try {
		    while (true) {
		        // 100 是超时时间（ms），在该时间内 poll 会等待服务器返回数据
		    	ConsumerRecords<Long, String> records = consumer2.poll(Duration.ofMillis(Long.MAX_VALUE)); 
		        // poll 返回一个记录列表。
		        // 每条记录都包含了记录所属主题的信息、记录所在分区的信息、记录在分区里的偏移量，以及记录的键值对。
		        for (ConsumerRecord<Long, String> record : records) {
		        	KafkaConsumerService.revokeOrder(record.value());
			    }
		        /*consumer.commitSync();//同步*/
		    } 
		} finally {
		    // 关闭消费者,网络连接和 socket 也会随之关闭，并立即触发一次再均衡
			consumer2.close();
		}
    }
	
	/**
	 * 消费者消费
	 * @param consumer3
	 */
	public static void orderChangeConsumer(KafkaConsumer<Long, String> consumer3){
    	try {
		    while (true) {
		        // 100 是超时时间（ms），在该时间内 poll 会等待服务器返回数据
		    	ConsumerRecords<Long, String> records = consumer3.poll(Duration.ofMillis(Long.MAX_VALUE)); 
		        // poll 返回一个记录列表。
		        // 每条记录都包含了记录所属主题的信息、记录所在分区的信息、记录在分区里的偏移量，以及记录的键值对。
		        for (ConsumerRecord<Long, String> record : records) {
		        	KafkaConsumerService.orderChange(record.value());
			    }
		        /*consumer.commitSync();//同步*/
		    } 
		} finally {
		    // 关闭消费者,网络连接和 socket 也会随之关闭，并立即触发一次再均衡
			consumer3.close();
		}
    }
	

	/**
	 * 消费者消费
	 * @param consumer4
	 */
	public static void tradeDepthConsumer(KafkaConsumer<Long, String> consumer4){
    	try {
    		Set<TopicPartition> assignment = new HashSet<>();
        	// 在poll()方法内部执行分区分配逻辑，该循环确保分区已被分配。
        	// 当分区消息为0时进入此循环，如果不为0，则说明已经成功分配到了分区。
        	while (assignment.size() == 0) {
        		consumer4.poll(Duration.ofMillis(1000));
        	    // assignment()方法是用来获取消费者所分配到的分区消息的
        	    // assignment的值为：topic-demo-3, topic-demo-0, topic-demo-2, topic-demo-1
        	    assignment = consumer4.assignment();
        	}
        	// 指定分区从末尾消费
        	consumer4.seekToEnd(assignment);
        	
		    while (true) {
		        // 100 是超时时间（ms），在该时间内 poll 会等待服务器返回数据
		    	ConsumerRecords<Long, String> records = consumer4.poll(Duration.ofMillis(1000)); 
		        // poll 返回一个记录列表。
		        // 每条记录都包含了记录所属主题的信息、记录所在分区的信息、记录在分区里的偏移量，以及记录的键值对。
		        for (ConsumerRecord<Long, String> record : records) {
		        	KafkaConsumerService.tradeDepth(record.value());
			    }
		        /*consumer.commitSync();//同步*/
		    } 
		} finally {
		    // 关闭消费者,网络连接和 socket 也会随之关闭，并立即触发一次再均衡
			consumer4.close();
		}
    }
}
